package com.pig4cloud.pigx.contract.websocket.subscribe;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.pig4cloud.pigx.contract.enums.ContractOrderEnum;
import com.pig4cloud.pigx.contract.mongo.entity.ContractSymbolMongo;
import com.pig4cloud.pigx.contract.mq.listener.DepthListener;
import com.pig4cloud.pigx.contract.vo.AbstractTopicVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 */
@Slf4j
@Component
public class DepthMap extends DataListAbstractTopic {
	protected String TOPICNAME = "depthMap";
	@Autowired
	private RedisTemplate redis;
	@Autowired
	private DepthListener depthListenerl;
	private ThreadPoolTaskExecutor websocketThreadPool;

	@Autowired
	private MongoTemplate mongo;

	private PushThread pushThread = new PushThread();

	public DepthMap(ThreadPoolTaskExecutor websocketThreadPool) {
		this.websocketThreadPool = websocketThreadPool;
		init();
	}

	private void init() {
		// 开启线程，发送数据
		websocketThreadPool.execute(pushThread);
		threadCheck();
	}

	private TreeMap<BigDecimal, List<BigDecimal>> getDepthFromRedis(String symbol, String side, String level) {
		TreeMap<BigDecimal, List<BigDecimal>> depTreeMap = new TreeMap<BigDecimal, List<BigDecimal>>();

		try {
			Object o = redis.opsForValue().get("depth:" + symbol + ":" + side + ":" + level);
			if (o != null) {
				JSONObject jsonObject = JSONObject.parseObject(o.toString());
				Gson gson = new Gson();
				depTreeMap = gson.fromJson(jsonObject.toJSONString(), new TypeToken<TreeMap<BigDecimal, List<BigDecimal>>>() {
				}.getType());


			}
//			if(depMap != null){
//				depTreeMap.putAll(depMap);
//			}

		} catch (Exception ignore) {
		}

		return depTreeMap;

	}

	private TreeMap<BigDecimal, BigDecimal> getFromRedis(String symbol, String level, String side) {

		Object redisObj = null;
		if (side.equals(ContractOrderEnum.Side.MORE.getValue())) {
			redisObj = redis.opsForValue().get("depth:" + symbol + ":more");
		} else {
			redisObj = redis.opsForValue().get("depth:" + symbol + ":empty");
		}
		if (redisObj != null) {
			JSONObject jsonObject = JSONObject.parseObject(redisObj.toString());

			String depthString = jsonObject.get(level).toString();
			Gson gson = new Gson();
			TreeMap<BigDecimal, BigDecimal> m = gson.fromJson(depthString, new TypeToken<TreeMap<BigDecimal, BigDecimal>>() {
			}.getType());

			return m;
		}
		return null;
	}

	private List<TreeMap<BigDecimal, List<BigDecimal>>> getFromDepthListenerl(String symbol, String level, String side) {

		BigDecimal lastprice = new BigDecimal(redis.opsForValue().get(symbol + "_last_price").toString());
		return depthListenerl.getDepth(symbol, lastprice, level, side);
	}


	private void threadCheck() {
		ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
		service.scheduleAtFixedRate(() -> {
			try {
				if (pushThread.getState() == Thread.State.TERMINATED) {
					// 重新开启线程，发送数据
					pushThread = new PushThread();
					websocketThreadPool.execute(pushThread);
				}

			} catch (Exception e) {
				log.info("check depth thread error" + e.getMessage());
			}

		}, 10000, 10000, TimeUnit.MILLISECONDS);
	}

	public class PushThread extends Thread {
		@Override
		public void run() {
			while (true) {
				try {
					long startTime = System.currentTimeMillis();

					for (Map.Entry<WebSocketSession, AbstractTopicVO> entry : subList.entrySet()) {
						try {
							if (!entry.getKey().isOpen()) {
								subList.remove(entry.getKey());
								log.info("ws 移除已关闭的session");
								continue;
							}

							AbstractTopicVO param = entry.getValue();
							List<String> dataList = param.getDataList();
							String symbol = dataList.get(1);
							String level = dataList.get(3);
							// 如果没有传深度，则使用最小深度
							if ("".equals(level)) {
								ContractSymbolMongo mongoOne = mongo.findOne(Query.query(Criteria.where("name").is(symbol)), ContractSymbolMongo.class);
								assert mongoOne != null;
								Integer accuracy = mongoOne.getMainCurAccuracy();
								BigDecimal l = BigDecimal.ONE;
								for (int i = accuracy; i > 0; i--) {
									l = l.divide(BigDecimal.TEN, accuracy, RoundingMode.HALF_UP);
								}
								level = l.toPlainString();
							}

							TreeMap<BigDecimal, List<BigDecimal>> moreTreeMap = getDepthFromRedis(symbol, ContractOrderEnum.Side.MORE.getValue(), level);
							TreeMap<BigDecimal, List<BigDecimal>> emptyTreeMap = getDepthFromRedis(symbol, ContractOrderEnum.Side.EMPTY.getValue(), level);

							NavigableMap<BigDecimal, List<BigDecimal>> moreMap = moreTreeMap.descendingMap();
							NavigableMap<BigDecimal, List<BigDecimal>> emptyMap = emptyTreeMap.descendingMap();


							JSONObject depthJsonObject = new JSONObject();
							JSONArray jsonArray_more = new JSONArray();
							final BigDecimal t = moreMap.get(moreMap.lastKey()).get(1);
							moreMap.forEach((k, v) -> {
								JSONArray arr = new JSONArray();
								arr.add(k);
								arr.add(v.get(0));
								arr.add(v.get(1));
								arr.add(v.get(1).divide(t, 4, RoundingMode.DOWN));

								jsonArray_more.add(arr);
							});
							depthJsonObject.put(ContractOrderEnum.Side.MORE.getValue(), jsonArray_more);


							JSONArray jsonArray_empty = new JSONArray();
							final BigDecimal tm = emptyMap.get(emptyMap.firstKey()).get(1);
							emptyMap.forEach((k, v) -> {
								JSONArray arr = new JSONArray();
								arr.add(k);
								arr.add(v.get(0));
								arr.add(v.get(1));
								arr.add(v.get(1).divide(tm, 4, RoundingMode.DOWN));
								jsonArray_empty.add(arr);
							});

							depthJsonObject.put(ContractOrderEnum.Side.EMPTY.getValue(), jsonArray_empty);

							entry.getKey().sendMessage(getTextMessage(param.getDataList(), depthJsonObject));

						} catch (Throwable e) {
							e.printStackTrace();
							log.error("Depth push thread error" + e.getMessage());
						}
					}

					long end = System.currentTimeMillis() - startTime;
					log.info("-------------------depth wsocket push ------------------------------");
					log.info("--------------------------------------------------------------------");
					log.info("--------------------------" + end + "------------------------------------------");
					log.info("--------------------------------------------------------------------");
					try {
						Thread.sleep(500);
					} catch (Exception ignored) {
						log.info("--------------depth wsocket push sleep ERROR--------------");
					}
				} catch (Exception e) {
					e.printStackTrace();
					log.error("Depth push thread error" + e.getMessage());
				}
			}
		}
	}
}
